package com.nx.platform.es.biz.esspider.source;


import com.google.common.base.Preconditions;
import com.google.common.primitives.Longs;
import com.google.common.util.concurrent.AbstractExecutionThreadService;
import com.nx.platform.es.biz.esspider.deliver.Deliverer;
import com.nx.platform.es.biz.esspider.entity.Message;
import com.nx.platform.es.common.utils.MoreFunctions;
import com.nx.platform.es.common.utils.MoreSplitters;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;

/**
 * @author
 * @since 2017年3月24日
 */
public class FileSourceService extends AbstractExecutionThreadService implements SourceService {

    private static final Logger LOGGER = LogManager.getLogger(FileSourceService.class);

    private final Deliverer deliverer;
    private final String[] fileNames;

    private volatile boolean flag = true;

    public FileSourceService(Deliverer deliverer, String fileNames) {
        this.deliverer = deliverer;
        this.fileNames = MoreFunctions.toStringArray(fileNames, MoreSplitters.COMMA);
    }

    @Override
    protected void triggerShutdown() {
        flag = false;
    }

    @Override
    protected void run() throws Exception {
        Preconditions.checkNotNull(deliverer);
        for (String fileName : fileNames) {
            if (fileName.length() == 0) {
                continue;
            }
            BufferedReader in = null;
            try {
                in = new BufferedReader(new InputStreamReader(new FileInputStream(fileName)));
                for (String line = in.readLine(); line != null && flag; line = in.readLine()) {
                    line = line.trim();
                    if (line.length() > 2) {
                        deliverer.deliver(parse(line));
                    }
                }
            } finally {
                if (in != null) {
                    in.close();
                }
            }
        }
    }

    private Message parse(String line) {
        String[] arr = MoreSplitters.COMMA.split(line);
        if (arr.length < 2) {
            LOGGER.warn("Message ignored, {}", line);
            return null;
        }
        String type = arr[0].trim();
        Long id = Longs.tryParse(arr[1].trim());
        if (id == null) {
            LOGGER.warn("Message ignored, {}", line);
            return null;
        }
        return new Message(this.getClass(), type, id, null, line);
    }

}
